package com.eva.biz.system.impl;

import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.conditions.update.UpdateWrapper;
import com.eva.biz.system.SystemJobTriggerBiz;
import com.eva.biz.system.dto.TriggerJobDTO;
import com.eva.core.constants.Constants;
import com.eva.core.exception.LockedException;
import com.eva.core.job.BaseDistributer;
import com.eva.core.job.BaseJob;
import com.eva.core.job.JobContext;
import com.eva.core.job.JobParam;
import com.eva.core.utils.Utils;
import com.eva.dao.system.SystemJobMapper;
import com.eva.dao.system.SystemJobSnippetMapper;
import com.eva.dao.system.model.SystemJob;
import com.eva.dao.system.model.SystemJobLog;
import com.eva.dao.system.model.SystemJobSnippet;
import com.eva.service.system.SystemJobLogService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.RandomStringUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.Date;
import java.util.List;
import java.util.UUID;

@Slf4j
@Service
public class SystemJobTriggerBizImpl implements SystemJobTriggerBiz {

    @Autowired
    private SystemJobLogService systemJobLogService;

    @Autowired
    private SystemJobMapper systemJobMapper;

    @Autowired
    private SystemJobSnippetMapper systemJobSnippetMapper;

    private ThreadLocal<String> jobUnique = new ThreadLocal<>();

    /**
     * 分成三种情况
     * 1. 无需分片&无需保证时间节点不丢失的任务：直接执行任务
     * 2. 需分片的任务：执行分片任务，并将分片记录进SYSTEM_JOB_SNIPPET。
     * 3. 需保证时间节点不丢失的任务：产生一条记录
     */
    /**
     * 定时任务分为2种情况
     * 1. 按照一定的时间间隔定时执行，服务器时间不一致，例如A服务器为10：00：00，B服务器为10：00：10，按照每15秒钟执行一次，则
     *   A服务器的执行时间节点有；10：00：15，10：00：30，10：00：45
     *   B服务器的执行时间节点有：10：00：25，10：00：40，10；00：55
     *   结果：时间间隔无法保证
     * 那么任务的执行间隔不再固定的是15秒。
     * 2. 按照指定时间点执行，例如每天4点执行。此时如果AB服务器时间如上说明，则
     *   始终都是B服务器先抵达4点。如果处理的数据是为今日下单的用户发送发货短信，那么A服务器始终都无法处理到数据。
     *   结果：A资源浪费
     * 解决办法：
     * 每次任务执行时做两件事情
     * 1. 验证当前业务时间是否大于任务预计下一次执行时间，如果是，则可以执行
     * 2. 执行完任务后，修改任务的预计下一次执行时间。
     * 这样处理后，以上2种情况将发生以下变化
     * 1. 按照一定的时间间隔定时执行，则
     *   A服务器率先获得任务锁，执行了任务
     *   A服务器10：00：00执行（预计下次执行时间为10：00：15）
     *   B服务器10：00：10执行（此次执行跳过，未获得到任务锁）
     *   A服务器10：00：15执行（预计下次执行时间为10：00：30）
     *   B服务器10：00：25执行（此次执行跳过，未到达下一次执行时间10：00：30）
     *   此时A服务器宕机，原本10：00：30的没有得到处理
     *   B服务器10：00：40执行（预计下次执行时间为10：00：55）
     *   此时A服务器开启，启动为10：00：42，此时B服务器时间为10：00：52
     *   B服务器10：00：55执行（预计下次执行时间为10：01：10）
     *   A服务器10：00：57执行（此次执行跳过，未达到下一次执行时间10：01：10）
     *   B服务器10：01：10执行（预计下次执行时间为10：01：25）
     *   A服务器10：01：12执行（此次执行跳过，未达到下一次执行时间10：01：25）
     *   这样，不仅保证了执行间隔一致，在服务器宕机时，其他机器在接收定时任务后继续保持间隔。但这样会出现始终都是一台机器在处理数据的情况。
     * 2. 按照指定时间点执行，例如每天4点执行，则
     *   B服务器率先抵达4点，执行任务后，任务的预计的下一次执行时间节点为20号4点
     *   A服务器抵达4点，由于执行时间未到达20号的4点，此次执行跳过。
     *   B服务器宕机
     *   A服务器抵达20号的4点，接手任务执行。预计下一次执行时间为21号的4点
     *   B服务器开启，此时时间为21号5点
     *   B服务器率先抵达21号4点，继续接手任务执行，预计下一次执行时间为22号的4点
     *   A服务器抵达21号的4点，由于执行时间未到达22号的4点，此次执行跳过
     *   到此B服务器宕机重启后又重新接管任务的执行。但这样依然会出现始终都是一台机器在处理数据的情况。
     *
     * 解决办法
     * 增加任务分片，任务每次执行都插入一条分片记录，各服务器上的分片扫描器每秒钟扫描分片记录并执行
     * 1. 按照一定的时间间隔定时执行，则
     *   A服务器率先获得任务锁，执行了任务
     *   A服务器10：00：00执行（产生了一个分片，该分片可能被A或B服务器执行，预计下次执行时间为10：00：15）
     *   B服务器10：00：10执行（此次执行跳过，未获得到任务锁）
     *   A服务器10：00：15执行（产生了一个分片，该分片可能被A或B服务器执行，预计下次执行时间为10：00：30）
     *   B服务器10：00：25执行（此次执行跳过，未到达下一次执行时间10：00：30）
     *   此时AB服务器均被利用起来，一旦A服务器宕机，B服务器会接手来产生分片。
     * 2. 按照指定时间点执行，例如每天4点执行，则
     *   B服务器率先抵达4点，执行任务后，产生了任务分片，该分片可能被A或B服务器执行，任务的预计的下一次执行时间节点为下一天的4点
     *   A服务器抵达4点，由于执行时间未到达下一天的4点，此次执行跳过。
     *   此时AB服务器均被利用起来，一旦B服务器宕机，A服务器会接手来产生分片，B服务器重启后，B将重新接手产生分片。
     *
     * 总结：添加预计下一次执行时间节点后，分片始终由一台服务器X产生，并在服务器X宕机后其他服务器会接手。而分片始终由多台服务器处理，从而大到资源利用。
     *
     * 定时任务还需支持
     * 1、多服务器允许并发执行
     * @param dto
     */
    @Override
    public void trigger(TriggerJobDTO dto) {
        SystemJob job = null;
        JobParam jobParam = new JobParam();
        boolean locked = false;
        SystemJobLog systemJobLog = new SystemJobLog();
        try {
            Integer jobId = dto.getId();
            job = systemJobMapper.selectById(jobId);
            // 验证JOB
            boolean executable = this.checkJob(job, dto.getBusinessTime());
            if (!executable) {
                return;
            }
            // 初始化JOB数据
            jobUnique.set(RandomStringUtils.randomAlphabetic(8));
            log(job, "Job started");
            log(job, "Job thread name: {}", Thread.currentThread().getName());
            log(job, "Job scheduled fire time: {}", Utils.Date.format(dto.getScheduledFireTime()));
            log(job, "Job fire time: {}", Utils.Date.format(dto.getFireTime()));
            log(job, "Job business time: {}", Utils.Date.format(dto.getBusinessTime()));
            jobParam.setId(jobId);
            jobParam.setTriggerType(dto.getTriggerType());
            jobParam.setBusinessTime(dto.getBusinessTime());
            // 记录日志
            if (job.getWithLog()) {
                systemJobLog.setJobId(jobId);
                systemJobLog.setJobName(job.getJobName());
                systemJobLog.setRunTimeStart(new Date());
                systemJobLog.setServerIp(Utils.Server.getIP());
                systemJobLog.setBusinessTime(dto.getBusinessTime());
                systemJobLog.setStatus(Constants.Job.LogStatus.NONE.getCode());
                systemJobLog.setTriggerType(dto.getTriggerType());
                systemJobLog.setRemark("任务正在执行中");
                systemJobLogService.create(systemJobLog);
            }
            // 任务加锁，防止多服务器重复执行
            this.lock(job);
            locked = true;
            // 如果开启了异步执行，则创建一个分片
            if (job.getWithAsync()) {
                String snippetCode = this.createAsyncSnippet(job, jobParam);
                systemJobLog.setStatus(Constants.Job.LogStatus.SUCCESS.getCode());
                systemJobLog.setRemark("异步任务已转为分片任务，分片Code：" + snippetCode);
                return;
            }
            // 如果存在分片
            if (StringUtils.isNotBlank(job.getDistributeHandler())) {
                int snippetCount = this.distribute(job, jobParam);
                systemJobLog.setStatus(Constants.Job.LogStatus.SUCCESS.getCode());
                systemJobLog.setRemark("任务已分发成" + snippetCount + "个分片任务");
                return;
            }
            // 不存在分片，直接执行任务（根据任务处理器名称获取任务处理器实例并执行）
            BaseJob jobBean = Utils.SpringContext.getBean(job.getHandler(), BaseJob.class);
            JobContext jobContext = jobBean.execute(jobParam);
            // 解锁任务
            this.unlock(job, dto.getNextFireTime());
            locked = false;
            // 修改日志信息
            if (job.getWithLog()) {
                systemJobLog.setRemark("任务执行完成");
                systemJobLog.setHandleTotalSize(jobContext.getHandleTotalSize());
                systemJobLog.setHandleSuccessSize(jobContext.getHandleSuccessSize());
                systemJobLog.setContext(jobContext.getContext());
                systemJobLog.setStatus(Constants.Job.LogStatus.SUCCESS.getCode());
            }
        } catch (LockedException e) {
            log(job, e.getMessage());
            // 修改日志状态
            if (job.getWithLog()) {
                systemJobLog.setStatus(Constants.Job.LogStatus.IGNORE.getCode());
                systemJobLog.setRemark("任务正在被其他服务执行或锁定");
            }
        } catch (Exception e) {
            if (job != null) {
                error(job, e);
                // 修改日志状态
                if (job.getWithLog()) {
                    systemJobLog.setStatus(Constants.Job.LogStatus.FAILED.getCode());
                    systemJobLog.setRemark("任务执行失败，出现异常：" + e.getMessage());
                }
            } else {
                e.printStackTrace();
            }
        } finally {
            // 更新日志
            if (job != null && job.getWithLog() && systemJobLog.getId() != null) {
                systemJobLog.setRunTimeEnd(new Date());
                systemJobLog.setRunSpendTime(Integer.valueOf("" + (systemJobLog.getRunTimeEnd().getTime() - systemJobLog.getRunTimeStart().getTime())));
                systemJobLogService.updateById(systemJobLog);
            }
            // 任务解锁
            if (locked) {
                this.unlock(job, dto.getNextFireTime());
                log(job, "Job execute completed.");
            }
        }
    }

    /**
     * 锁定JOB
     * @param job JOB
     * @return
     */
    private void lock (SystemJob job) throws LockedException{
        log(job, "Lock job.");
        if (job.getStatus().equals(Constants.Job.JobStatus.RUNNING.getCode())) {
            throw new LockedException("EVA: job is running.");
        }
        SystemJob newJob = new SystemJob();
        newJob.setStatus(Constants.Job.JobStatus.RUNNING.getCode());
        newJob.setLockNumber(job.getLockNumber() + 1);
        // 不修改更新人和更新时间
        newJob.setUpdateUser(Constants.Ignore.IGNORE_USER);
        newJob.setUpdateTime(Constants.Ignore.IGNORE_TIME);
        int result = systemJobMapper.update(newJob, new UpdateWrapper<SystemJob>()
                .lambda()
                .eq(SystemJob::getId, job.getId())
                .eq(SystemJob::getLockNumber, job.getLockNumber())
        );
        if (result == 0) {
            throw new LockedException("EVA: job is locked.");
        }
        log(job, "Lock job completed.");
    }

    /**
     * 解锁JOB
     * @param job JOB
     * @return
     */
    private void unlock(SystemJob job, Date nextFireTime) {
        log(job, "Unlock job.");
        SystemJob newJob = new SystemJob();
        newJob.setId(job.getId());
        newJob.setStatus(Constants.Job.JobStatus.READY.getCode());
        newJob.setNextFireTime(nextFireTime);
        // 不修改更新人和更新时间
        newJob.setUpdateUser(Constants.Ignore.IGNORE_USER);
        newJob.setUpdateTime(Constants.Ignore.IGNORE_TIME);
        systemJobMapper.updateById(newJob);
        log(job, "Unlock job completed.");
    }

    /**
     * 创建异步分片
     * @param job JOB
     * @param jobParam JOB参数
     */
    private String createAsyncSnippet (SystemJob job, JobParam jobParam) {
        log(job, "Begin create async snippet.");
        String distributeGroup = UUID.randomUUID().toString();
        SystemJobSnippet snippet = new SystemJobSnippet();
        snippet.setJobId(job.getId());
        snippet.setJobName(job.getJobName());
        snippet.setJobBusinessTime(jobParam.getBusinessTime());
        snippet.setJobDistributeGroup(distributeGroup);
        snippet.setDistributeHandler(job.getDistributeHandler());
        snippet.setDistributeLimit(job.getDistributeLimit());
        snippet.setHandler(job.getHandler());
        snippet.setSnippetIndex(0);
        snippet.setSnippetCode(UUID.randomUUID().toString());
        snippet.setWithLog(job.getWithLog());
        snippet.setStatus(Constants.Job.JobStatus.READY.getCode());
        systemJobSnippetMapper.insert(snippet);
        log(job, "Create async snippet completed.");
        return snippet.getSnippetCode();
    }

    /**
     * 分发
     * @param job
     * @param jobParam
     * @return
     */
    private int distribute (SystemJob job, JobParam jobParam) {
        log(job, "Begin distribute.");
        String distributeGroup = UUID.randomUUID().toString();
        BaseDistributer distributer = Utils.SpringContext.getBean(job.getDistributeHandler(), BaseDistributer.class);
        List<List<?>> snippetDatas = distributer.getSnippetData(new BaseDistributer.DistributeDTO(job.getDistributeLimit()), jobParam);
        for (int i = 0; i < snippetDatas.size(); i++) {
            List snippetData = snippetDatas.get(i);
            SystemJobSnippet snippet = new SystemJobSnippet();
            snippet.setJobId(job.getId());
            snippet.setJobName(job.getJobName());
            snippet.setJobBusinessTime(jobParam.getBusinessTime());
            snippet.setJobDistributeGroup(distributeGroup);
            snippet.setHandler(job.getHandler());
            snippet.setSnippetIndex(i);
            snippet.setSnippetCode(UUID.randomUUID().toString());
            snippet.setSnippetData(JSON.toJSONString(snippetData));
            snippet.setSnippetDataSize(snippetData.size());
            snippet.setWithLog(job.getWithLog());
            snippet.setStatus(Constants.Job.JobStatus.READY.getCode());
            systemJobSnippetMapper.insert(snippet);
        }
        log(job, "distribute completed.");
        return snippetDatas.size();
    }

    /**
     * 验证Job是否可执行
     * @param job
     * @return
     */
    private boolean checkJob (SystemJob job, Date businessTime) {
        // 执行时间验证（任务预计下一次执行<=当前执行时间即可执行)
        if (businessTime != null && job.getNextFireTime() != null && job.getNextFireTime().compareTo(businessTime) == 1) {
            return Boolean.FALSE;
        }
        // 服务器验证
        if (StringUtils.isNotBlank(job.getAllowServerIps())) {
            String serverIp = Utils.Server.getIP();
            String[] ips = job.getAllowServerIps().split(",");
            boolean exists = false;
            for (String ip : ips) {
                if (ip.trim().equals(serverIp)) {
                    exists = true;
                    break;
                }
            }
            if (!exists) {
                return Boolean.FALSE;
            }
        }
        return Boolean.TRUE;
    }

    /**
     * 打印日志
     * @param job JOB
     * @param text 日志模版内容
     * @param params 日志参数
     */
    private void log (SystemJob job, String text, Object... params) {
        log.info("{}: " + text, String.format("%s_%s", job.getJobName(), jobUnique.get()), params);
    }

    /**
     * 打印错误日志
     * @param job JOB
     * @param e 异常对象
     */
    private void error (SystemJob job, Exception e) {
        log.error(String.format("EVA: %s_%s", job.getJobName(), jobUnique.get()) + ": Handle job throw an exception.", e);
    }
}
